package cn.trigram.concurrent.pool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用唯品会的代码
 * <br/>
 * From Tomcat 8.5.6, 传统的FixedThreadPool有Queue但线程数量不变，而CachedThreadPool线程数可变但没有Queue。
 * <br/>
 * Tomcat的线程池，通过控制TaskQueue，线程数，但线程数到达最大时会进入Queue中.
 * <br/>
 * 代码从Tomcat复制，主要修改包括：
 * <br/>
 * 1. 删除定期重启线程避免内存泄漏的功能
 * <br/>
 * 2. TaskQueue 中可能3次有锁的读取线程数量，改为只读取1次，这把锁也是这个实现里的唯一遗憾了
 * <br/>
 *
 * @see <a href="https://github.com/apache/tomcat/blob/trunk/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java">Tomcat
 * ThreadPoolExecutor</a>
 */
public final class QueuableCachedThreadPoolExecutor extends
        ThreadPoolExecutor {

  /**
   * 这个数字是指任务已提交但是还没完成的任务数。
   * 它包括了在队列中的任务和已经通过一个worker线程处理但还未开始执行的任务。
   * 这个数总是大于或等于 {@link #getActiveCount()} ()}
   * <br/>
   * <h2>原文</h2>
   * This includes tasks in the queue and tasks
   * that have been handed to a worker thread
   * but the latter did not start executing the task yet.
   * This number is always greater or equal to {@link #getActiveCount()}.
   */
  private final AtomicInteger submittedCount = new AtomicInteger(0);

  public QueuableCachedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                                          TimeUnit unit,
                                          ControllableQueue workQueue, ThreadFactory threadFactory,
                                          RejectedExecutionHandler handler
  ) {

    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    workQueue.setExecutor(this);
    prestartAllCoreThreads(); // NOSOANR
  }

  @Override
  protected void afterExecute(Runnable r, Throwable t) {

    submittedCount.decrementAndGet();
  }

  public int getSubmittedCount() {

    return submittedCount.get();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void execute(Runnable command) {

    execute(command, 0, TimeUnit.MILLISECONDS);
  }

  /**
   * 在未来的某个时间执行给定的任务。这个任务可能会在新的线程或者一个线程池中又或者在调用线程中，由Executor决定。
   * 如果没有线程可用，任务会被添加进工作队列中。如果这个工作队列满了，系统会等待指定的时间，
   * 如果在等待时间超过后队列仍然是满的，这个方法就会抛出一个 {@link RejectedExecutionException}
   * <h3>原文：</h3>
   * Executes the given command at some time in the future. The command may execute in a new thread,
   * in a pooled
   * thread, or in the calling thread, at the discretion of the <tt>Executor</tt> implementation. If
   * no threads are
   * available, it will be added to the work queue. If the work queue is full, the system will wait
   * for the specified
   * time and it throw a RejectedExecutionException if the queue is still full after that.
   *
   * @param command
   *     the runnable task
   * @param timeout
   *     A timeout for the completion of the task
   * @param unit
   *     The timeout time unit
   * @throws RejectedExecutionException
   *     if this task cannot be accepted for execution - the queue is full
   * @throws NullPointerException
   *     if command or unit is null
   */
  public void execute(Runnable command, long timeout, TimeUnit unit) {

    submittedCount.incrementAndGet();
    try {
      super.execute(command);
    } catch (RejectedExecutionException rx) { // NOSONAR
      // not to re-throw this exception because this is only used to find out whether the pool is full, not for a
      // exception purpose
      final ControllableQueue queue = (ControllableQueue) super.getQueue();
      try {
        if (!queue.force(command, timeout, unit)) {
          submittedCount.decrementAndGet();
          throw new RejectedExecutionException("Queue capacity is full.");
        }
      } catch (InterruptedException ignore) {
        submittedCount.decrementAndGet();
        throw new RejectedExecutionException(ignore);
      }
    }
  }

  /**
   * 这里去除了原来的tomcat6 下的corePoolSize 必须设置为0 的代码。
   * <br/>
   * 其实就是一个在最大线程已满的时候，将任务放进队列中，这是使用的默认容量是100
   * <br/>
   * 此队列尽可能的让线程池先使用最大的线程数处理任务
   * <br/>
   *
   * @see <a href="https://github.com/apache/tomcat/blob/trunk/java/org/apache/tomcat/util/threads/TaskQueue.java">Tomcat
   * TaskQueue</a>
   */
  protected static class ControllableQueue extends LinkedBlockingQueue<Runnable> {

    private static final long serialVersionUID = 5044057462066661171L;

    private transient volatile QueuableCachedThreadPoolExecutor executor = null;

    public ControllableQueue(int capacity) {

      super(capacity);
    }

    public void setExecutor(QueuableCachedThreadPoolExecutor tp) {

      executor = tp;
    }

    public boolean force(Runnable o) {

      if (executor.isShutdown()) {
        throw new RejectedExecutionException(
            "Executor not running, can't force a command into the queue");
      }
      return super.offer(o); // forces the item onto the queue, to be used if the task is rejected
    }

    /**
     * 在给定的超时时间内尝试强行提交任务
     */
    public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {

      if (executor.isShutdown()) {
        throw new RejectedExecutionException(
            "Executor not running, can't force a command into the queue");
      }
      // forces the item onto the queue, to be used if the task is rejected
      return super.offer(o, timeout, unit);
    }

    /**
     * 提交任务，有以下判断顺序：
     * <ol>
     *   <li>当前可用线程数大于等于最大线程数，任务提交到队列中</li>
     *   <li>线程池中的已提交数小于当前可用线程数，代表有空闲的线程，但是也是提交到队列等待线程自动从队列获取任务处理</li>
     *   <li>当前可用线程数小于最大线程数时，应该直接创建一个新的线程</li>
     *   <li>其它情况直接提交队列</li>
     * </ol>
     */
    @Override
    public boolean offer(Runnable o) {
      // spring-side: threadPool.getPoolSize() 是个有锁的操作，所以尽量减少
      // 在tomcat中，这里调用了三次getPoolSize 方法

      int currentPoolSize = executor.getPoolSize();

      // 当前可用线程数大于等于最大线程数，任务提交到队列中
      if (currentPoolSize >= executor.getMaximumPoolSize()) {
        return super.offer(o);
      }
      // 有空闲线程并且当前线程数已经达到最大线程数，任务提交到队列中
      if (executor.getSubmittedCount() < currentPoolSize) {
        return super.offer(o);
      }
      // 当前线程数少于最大线程数，强制创建线程
      if (currentPoolSize < executor.getMaximumPoolSize()) {
        return false;
      }
      // 最后的保险，直接提交队列
      return super.offer(o);
    }

  }

}
